/*******************************************************************************
* COPYRIGHT LIPUJUN TECH CO.,LTD
********************************************************************************
* 文件名称:  mqttCom.c
* 功能描述:
* 使用说明:  无
* 文件作者:  LIPUJUN
* 编写日期:  2019/4/25
* 修改历史:
* 修改版本  修改日期       修改人   修改内容
* ------------------------------------------------------------------------------
  01a       2019/4/25       李普军   创建
*******************************************************************************/

/******************************* 包含文件声明 ********************************/
#include <sys/select.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include "link.h"
#include "common.h"
#include "user_core.h"
#include "msg_core.h"
#include <sys/ioctl.h>
#include <linux/sockios.h>

extern AIR_APP_INFO_T stMqttMsgInfo;
char *cmdtype[CMD_SIZE+1] = {"null","reg","sd","com","dc","upgrade","upgradeCom","h","onoff","unknown"};
char *g_cmdtype(int type)
{
    if(type >= 0 && type < CMD_SIZE)
        return cmdtype[type];
    else 
        return cmdtype[CMD_SIZE];
		
}

INT32 create_client(const CHAR* ip, INT16 port)
{
    INT32    sockfd;  
    static int lastT = 0;
    int curT = currentTimeGet_int();
    struct sockaddr_in    servaddr;
	struct linger so_linger = { 1, 0 };
    INT32 keepAlive = 1; // 开启keepalive属性
    INT32 keepIdle = stMqttMsgInfo.keepIdle ;// 如该连接在60秒内没有任何数据往来,则进行探测 
    INT32 keepInterval = stMqttMsgInfo.keepInterval; // 探测时发包的时间间隔为5 秒
    INT32 keepCount = stMqttMsgInfo.keepCount; // 探测尝试的次数.如果第1次探测包就收到响应了,则后2次的不再发.  

    if( (sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0){  
    	Error("Create socket error: %s (%d)", strerror(errno),errno);  
    	return -1;
    } 
  
    memset(&servaddr, 0, sizeof(servaddr));  
    servaddr.sin_family = AF_INET;  
    servaddr.sin_port = htons(port);  
    if( inet_pton(AF_INET, ip, &servaddr.sin_addr) <= 0){  
	   Error("inet_pton error for server ip:%s",ip);
	   close(sockfd); 
       return -1; 
    }   
  
    if( connect(sockfd, (struct sockaddr*)&servaddr, sizeof(servaddr)) < 0){ 
        close(sockfd);
        if((curT - lastT) >= 300){
            Error("Connect to %s:%d Error: %s (%d)",ip,port,strerror(errno),errno); 	
            lastT = curT;	
        }
        return -1;  
    }
    //lastT = curT;
    if(setsockopt(sockfd, SOL_SOCKET, SO_KEEPALIVE, (void *)&keepAlive, sizeof(keepAlive)) < 0){
	    Error("set socket opt keep alive %d Error: %s (%d)",keepAlive,strerror(errno),errno); 
	    close(sockfd);
	    return -1;
    }
    if(setsockopt(sockfd, SOL_TCP, TCP_KEEPIDLE, (void*)&keepIdle, sizeof(keepIdle)) < 0){
	    Error("set socket opt keep Idle %d Error: %s (%d)",keepIdle,strerror(errno),errno); 
	    close(sockfd);
	    return -1;
    }
    if(setsockopt(sockfd, SOL_TCP, TCP_KEEPINTVL, (void *)&keepInterval, sizeof(keepInterval)) < 0){
	    Error("set socket opt keep Interval %d Error: %s (%d)",keepInterval,strerror(errno),errno); 
	    close(sockfd);
	    return -1;
    }
    if( 0 != setsockopt(sockfd, SOL_SOCKET, SO_LINGER, &so_linger, sizeof(so_linger))){
        Error( "set socket opt force close Error: %s (%d)", strerror(errno), errno );
    }
    if(setsockopt(sockfd, SOL_TCP, TCP_KEEPCNT, (void *)&keepCount, sizeof(keepCount)) < 0){
        Error("set socket opt keep Count %d Error: %s (%d)",keepCount,strerror(errno),errno); 
        close(sockfd);
        return -1;
    }
    if(fcntl(sockfd, F_SETFL, fcntl(sockfd, F_GETFL) | O_NONBLOCK) < 0){
        Error("set fcntl socket O_NONBLOCK Error: %s (%d)",strerror(errno),errno); 
        close(sockfd);
        return -1;
    }
    INFO ("TCP Connect %s:%d OK.", ip , port);
    return sockfd;
}
INT32 socket_connect(void)
{
	stMqttMsgInfo.sock = create_client(stMqttMsgInfo.serverIp , stMqttMsgInfo.port);
	if(stMqttMsgInfo.sock > 0){
		INFO ("TCP Connect IOT_STATUS_CONNECT OK.");
		stMqttMsgInfo.link = IOT_STATUS_CONNECT;
		return stMqttMsgInfo.sock;
	}
	return -1;
}
INT32 socket_disconnect(void)
{
	if(stMqttMsgInfo.sock > 0){
		close(stMqttMsgInfo.sock);
		stMqttMsgInfo.sock = 0;
		INFO ("Disconnect Network.");
	}
	return OK;	
}

INT32 link_sendMsg(void* msg)
{
	UINT8 msg_buf[MSG_QUEUE_DATA_LEN] = {0};
	MSG_HEADER_T msg_header = {0};
	MSG_DATA_T *data = (MSG_DATA_T *)msg;
	INT32 len = 0;
	INT32 pos = 0;
	UINT32 size = 0;

	
	if(data == NULL){
	    Error("link send data == NULL");
	    return ERROR;
	}
	if(stMqttMsgInfo.link != IOT_STATUS_CONNECT){
	    Error("link send stMqttMsgInfo.link != IOT_STATUS_CONNECT");
	    return 0;
 	}
  	
	msg_header.head = htons(_MSG_HEADER_);
	if(APP_DEVICE_ID_SIZE_V2 == strlen(stMqttMsgInfo.devId))
	    msg_header.ver = APP_SUPPORT_MSG_VER_V2;
	else
	    msg_header.ver = APP_SUPPORT_MSG_VER;
	
	msg_header.type = data->type;
	msg_header.size = htons(data->len);
	memcpy(msg_header.id,stMqttMsgInfo.devId,sizeof(msg_header.id));
	
	memcpy(msg_buf,&msg_header,MSG_HEADER_SIZE);
	memcpy(&msg_buf[MSG_HEADER_SIZE],data->data,data->len);

	size = data->len + MSG_HEADER_SIZE;
	while(pos < size){
	    len = send(stMqttMsgInfo.sock,&msg_buf[pos],size - pos,0);
	    if(len <= 0){
		Error("link send data(%d) error:%s(errno:%d)",len,strerror(errno),errno);
		stMqttMsgInfo.link = IOT_STATUS_DROP;
		return 0;
	    }
	    pos += len;
	}

	INFO("(DEV->HOST) send msg:[%s,<<%s>>:%d]<%s>",stMqttMsgInfo.devId,g_cmdtype(data->type),size,data->data);       
	return size;
}
void check_socket(void)
{
    switch (stMqttMsgInfo.link)
    {
        case IOT_STATUS_LOGIN:
            if(stMqttMsgInfo.poeIsON)
                socket_connect();           
            break;	   
        case IOT_STATUS_CONNECT:     
            break;
        case IOT_STATUS_DROP:
            Error ("SOCKET IOT_STATUS_DROP !");
            socket_disconnect();            
            stMqttMsgInfo.link = IOT_STATUS_LOGIN;
	    usleep(1000);//500ms
            break;

        default:
            Error ("check_socket, default!");
            break;
    }
}

INT32 msg_check(MSG_HEADER_T *msg_header)
{
	if(ntohs(msg_header->head) != _MSG_HEADER_){
		Error("message check : header error %x",msg_header->head);
		return ERROR;
	}
	if(msg_header->type == CMD_NULL || msg_header->type >= CMD_SIZE){
		Error("message check : type error %d", msg_header->type);
		return ERROR;
	}
	if(0 != strncmp((const char*)msg_header->id, (const char*)stMqttMsgInfo.devId,strlen(stMqttMsgInfo.devId))){
		Error("message check : client Id error %s != %s",stMqttMsgInfo.devId , msg_header->id);
		return ERROR;
	}
	return OK;
}
INT32 comunicationLink_task (void *arg)
{    
    struct timeval timeout={0,200000}; //200ms timeout
    fd_set rfds;
    UINT32 index = 0;
    MSG_HEADER_T *msg_header = NULL;
    UINT8 msg_buf[MSG_QUEUE_DATA_LEN];
    UINT8 type = CMD_NULL;
    UINT8 flg = 0;
    UINT16 msg_size = 0;
    UINT8 ver = APP_SUPPORT_MSG_VER;
    check_socket();	     
    memset(msg_buf,0,sizeof(msg_buf));
	
    while (1)
    {
	INT32 ret;
	INT32 sockfd = stMqttMsgInfo.sock;
	INT32 len = 0;

        timeout.tv_sec = 0;
        timeout.tv_usec = 200000;//200ms

	usleep(50000);//50ms
        if(stMqttMsgInfo.link != IOT_STATUS_CONNECT){//维持服务器稳定，断开重连
	    check_socket();
            memset(msg_buf,0,sizeof(msg_buf));
            flg = 0;
            index = 0;
            msg_size = 0;
	    continue;
	}
	if(flg == 0){//帧头
	    if(index == MSG_HEADER_SIZE){
		msg_header = (MSG_HEADER_T *)msg_buf;	
		if(msg_check(msg_header) == OK){
		    //INFO("header: head[%x],type[%s],ver[%x],id[%s],size[%d]",ntohs(msg_header->head),g_cmdtype(msg_header->type),msg_header->ver,msg_header->id,ntohs(msg_header->size));
		    type = msg_header->type;
                    ver = msg_header->ver;
		    msg_size = ntohs(msg_header->size);
		    memcpy(msg_buf,&msg_buf[index], index - MSG_HEADER_SIZE);
		    flg = 1;
		    index = index - MSG_HEADER_SIZE;                    
		}else{//!= OK
		    memcpy(msg_buf,&msg_buf[1], MSG_HEADER_SIZE - 1);
		    index--;
		}
	    }
	}else {//数据
        if(msg_size == index){
            msg_buf[msg_size] = 0;
            if(CMD_UPLOAD_FILE == type && msgBdRecvUniq(type)){
                INFO("(HOST->DEV) repeat recv msg[%d]:[%s,<<%s>>:%d]<%s>",ver,stMqttMsgInfo.devId,g_cmdtype(type),msg_size+MSG_HEADER_SIZE,msg_buf);
                flg = 0;
                index = index - msg_size;
                continue;
            }
            air_rcv_msg (type, msg_buf, msg_size);
            INFO("(HOST->DEV) recv msg[%d]:[%s,<<%s>>:%d]<%s>",ver,stMqttMsgInfo.devId,g_cmdtype(type),msg_size+MSG_HEADER_SIZE,msg_buf);
            memset(msg_buf,0,msg_size + MSG_HEADER_SIZE);
            flg = 0;
		    index = index - msg_size;
	    }
	}
	

	mqttMsgTaskSend();      
	FD_ZERO(&rfds); //每次循环都要清空集合，否则不能检测描述符变化  
        FD_SET(sockfd,&rfds); //添加描述符        
       
	ret = select(sockfd+1,&rfds,NULL,NULL,&timeout);
	switch(ret)   //select使用  
        {  
            case -1: 
		Error("select -1 error:%s(errno:%d)\n",strerror(errno),errno);
		stMqttMsgInfo.link = IOT_STATUS_DROP;
		break; //select错误，退出程序  
            case 0:		
               //printf("select 0 error:%s(errno:%d)\n",strerror(errno),errno);
            break; //再次轮询  
            default:  
                if(FD_ISSET(sockfd,&rfds)) //测试sock是否可读，即是否网络上有数据  
                { 
			if(flg == 0 && index < MSG_HEADER_SIZE){//帧头
			    len = recv(sockfd, &msg_buf[index], MSG_HEADER_SIZE - index,0);
			    if(len <= 0){			    
				Error("header socket recv %d:%d error:%s(errno:%d)",index,len,strerror(errno),errno);
				stMqttMsgInfo.link = IOT_STATUS_DROP;
				continue;
			    }

			    index += len;
			}else if(flg && index < msg_size){//数据
			    len = recv(sockfd, &msg_buf[index], msg_size - index,0);
			    if(len <= 0){			    
				Error("data socket recv %d:%d error:%s(errno:%d)",index,len,strerror(errno),errno);
				stMqttMsgInfo.link = IOT_STATUS_DROP;
				continue;
			    }
			    index += len;
			}
             	}      
	     	break;
        }
    }

    return 0;
}

/******************************* 源文件结束 **********************************/
